package net.bwie.realtime.vehicle2.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.vehicle.bean.VehicleData;
import net.bwie.realtime.vehicle.function.GeoHashUdf;
import net.bwie.realtime.vehicle2.function.VehicleProvinceFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 车辆地理分布
 *
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class VehicleProvince {
    public static void main(String[] args) throws Exception {
        // 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源-source
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "vehicle-data-log");
        // 数据转换-transformation
        DataStream<String> processStream = ProcessLog(stringDataStream);
        processStream.print();
        // 数据输出-clickhouse
        JdbcUtil.sinkToClickhouseUpsert(processStream,
                "INSERT INTO vehicle_data.vehicle_province_count_report (geohash, vin_count) VALUES (?, ?)");
        // 触发执行-execute
        env.execute("VehicleProvince");
    }

    private static DataStream<String> ProcessLog(DataStream<String> stringDataStream) {
        SingleOutputStreamOperator<VehicleData> VehicleDataMap = stringDataStream.map(
                new MapFunction<String, VehicleData>() {
                    @Override
                    public VehicleData map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        String geoHash = GeoHashUdf.getGeoHash(jsonObject.getDouble("latitude"), jsonObject.getDouble("longitude"));
                        return new VehicleData(jsonObject.getString("vin"), jsonObject.getLong("timestamp"),
                                jsonObject.getDouble("longitude"), jsonObject.getDouble("latitude"),
                                jsonObject.getDouble("speed"), jsonObject.getDouble("mileage"),
                                jsonObject.getDouble("batteryLevel"), jsonObject.getDouble("batteryTemp"),
                                jsonObject.getDouble("motorTemp"), jsonObject.getString("chargingStatus"),
                                jsonObject.getDouble("energyConsumption"), jsonObject.getString("startTime"),
                                jsonObject.getString("endTime"), jsonObject.getDouble("batteryHealth"),
                                geoHash);
                    }
                }
        );
        SingleOutputStreamOperator<VehicleData> singleOutputStreamOperator = VehicleDataMap.assignTimestampsAndWatermarks(WatermarkStrategy
                .<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<VehicleData>() {
                            @Override
                            public long extractTimestamp(VehicleData element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }
                )
        );
        KeyedStream<VehicleData, String> vehicleDataStringKeyedStream = singleOutputStreamOperator.keyBy(o -> o.getGeohash());
        WindowedStream<VehicleData, String, TimeWindow> window = vehicleDataStringKeyedStream.window(
                TumblingEventTimeWindows.of(Time.seconds(30)));

        DataStream<String> vinStateStream = window.process(new VehicleProvinceFunction());
        return vinStateStream;
    }
}
